package com.neusoft.databus.agent.interactive.service;

import java.util.Date;

import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import com.neusoft.databus.agent.interactive.collect.CollectorProvider;
import com.neusoft.databus.agent.service.impl.CommandFlumeAgentService;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.exception.ConfigErrorException;
import com.neusoft.databus.common.exception.DisconnectException;
import com.neusoft.databus.common.exception.RetryableException;

import lombok.extern.slf4j.Slf4j;

/**
 * The entrance of collection task <br>
 * It will throw exception when any error occurred, or any configuration error.
 * The invoker must catch exception itself.
 *
 * @author sunchf
 * @date 2019年1月8日 上午9:52:43
 */
@Slf4j
@Service
public class CollectorService extends AbstractInteractiveService implements CollectorProvider {

    @Autowired
    private CommandFlumeAgentService commandFlumeAgentService;
    private final static int maxRetry = 1;

    @Override
    public CommandBean collect(final CommandBean commandBean) throws Exception {
        MDC.put("trace", commandBean.generateTraceId());
        this.doCollect(commandBean, 0);
        MDC.clear();
        return commandBean;
    }

    private void doCollect(CommandBean commandBean, int retry) throws Exception {
        final long startAt = System.currentTimeMillis();
        if (CollectorService.log.isDebugEnabled()) {
            CollectorService.log.debug("execute collect task");
        }
        if (retry == 0) {
            commandBean.setCollectBeginTime(new Date());
        }

        try {
            commandBean = this.getCollector(commandBean).collect(commandBean);
            commandBean.setCollectEndTime(new Date());
            if (commandBean.isAsync()) {
                this.commandFlumeAgentService.send(commandBean);
            }
        } catch (final Exception e) {
            if ((retry < CollectorService.maxRetry) && (e instanceof RetryableException)) {
                commandBean.setErrorCode(null);
                this.doCollect(commandBean, ++retry);
            } else {
                CollectorService.log.error("error occured when execute task", e);
                commandBean.setCollectEndTime(new Date());
                if (commandBean.isAsync()) {
                    if (e instanceof DisconnectException) {
                        commandBean.setErrorCode(DatabusErrorCode.ERR_CODE_406001);
                    } else if (e instanceof ConfigErrorException) {
                        commandBean.setErrorCode(DatabusErrorCode.ERR_CODE_406101);
                    } else {
                        commandBean.setErrorCode(DatabusErrorCode.ERR_CODE_400001);
                    }
                    commandBean.setErrorMessage(StringUtils.isEmpty(e.getMessage()) ? "未知异常" : e.getMessage());
                    this.commandFlumeAgentService.send(commandBean);
                } else {
                    throw e;
                }
            }
        } finally {
            CollectorService.log.info("collect service elapsed time:[{}]", System.currentTimeMillis() - startAt);
        }

    }

    @Override
    public CommandBean control(CommandBean commandBean) throws Exception {
        commandBean.setCollectBeginTime(new Date());
        commandBean = this.getCollector(commandBean).control(commandBean);
        commandBean.setCollectEndTime(new Date());
        return commandBean;
    }
}
